package org.example.marker.service.impl;

import org.example.marker.service.WorkerIdAssigner;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Wills
 * @date 2020/05/26
 */
@Slf4j
@Service
@EnableScheduling
public class DisposableWorkerIdAssigner implements WorkerIdAssigner {

    private static final String KEY_WORKID = "marker:workid";
    private static final String KEY_WORKID_INSTANCE = "marker:workid:instance:";

    @Autowired
    private BitsAllocator bitsAllocator;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private static final AtomicLong ATOMIC_LONG = new AtomicLong(0L);

    @Override
    public long assignWorkerId() {
        long worId = this.redisTemplate.opsForValue().increment(KEY_WORKID, 1);
        if (worId > this.bitsAllocator.getMaxWorkerId()) {
            this.redisTemplate.opsForValue().set(KEY_WORKID, "1");
            worId = 1L;
        }
        String existKey = KEY_WORKID_INSTANCE + worId;
        if (this.redisTemplate.hasKey(existKey)) {
            log.info("<<< workid [{}] has been assigned, try next one", worId);
            //retry if work id has been assigned
            return this.assignWorkerId();
        }
        this.redisTemplate.opsForValue().set(existKey, "DUMMY", 2, TimeUnit.MINUTES);
        ATOMIC_LONG.set(worId);
        return worId;
    }

    @Scheduled(cron = "0 * * * * ?")
    public void heartBeat() {
        String existKey = KEY_WORKID_INSTANCE + ATOMIC_LONG.get();
        log.info(">>> heart beat for workid key [{}]", existKey);
        this.redisTemplate.opsForValue().set(existKey, "DUMMY", 2, TimeUnit.MINUTES);
    }

}
